package net.opentsdb.tools;

import com.google.common.collect.Lists;
import com.google.common.util.concurrent.Atomics;
import com.stumbleupon.async.Deferred;
import net.opentsdb.core.TSDB;
import net.opentsdb.tsd.PipelineFactory;
import net.opentsdb.tsd.RpcManager;
import net.opentsdb.tsd.RpcPlugin;
import net.opentsdb.utils.Config;
import net.opentsdb.utils.Threads;
import org.jboss.netty.bootstrap.ServerBootstrap;
import org.jboss.netty.channel.socket.ServerSocketChannelFactory;
import org.jboss.netty.channel.socket.nio.NioServerBossPool;
import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;
import org.jboss.netty.channel.socket.nio.NioWorkerPool;
import org.jboss.netty.channel.socket.oio.OioServerSocketChannelFactory;
import org.jboss.netty.util.ThreadNameDeterminer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.file.FileSystem;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Map;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicReference;

/**
 * @author Hujf
 * @title: TSDMain
 * @date 2021-09-08 8:59
 * @description: opentsdb 启动类
 */
public class TSDMain {
    private static final short DEFAULT_FLUSH_INTERVAL = 1000;
    static Logger log = LoggerFactory.getLogger(TSDMain.class);

    /**
     * Reference to our singleton instance.  Set in {@link #initialize}.
     */
    private static final AtomicReference<RpcManager> INSTANCE = Atomics.newReference();

    /**
     * Prints usage and exits with the given retval.
     */
    static void usage(final ArgP argp, final String errmsg, final int retval) {
        log.error(errmsg);
        log.error("Usage: tsd --port=PORT"
                + " --staticroot=PATH --cachedir=PATH\n"
                + "Starts the TSD, the Time Series Daemon");
        if (argp != null) {
            log.info(argp.usage());
        }
        System.exit(retval);
    }

    private static TSDB tsdb = null;

    public static void main(String[] args) throws Exception {

        log.info("starting.");
        log.info(BuildData.revisionString());
        log.info(BuildData.buildString());

        try {
            System.in.close();
        } catch (IOException e) {
            log.warn("Failed to close stdin", e);
        }

        final ArgP argp = new ArgP();
        CliOptions.addCommon(argp);
        argp.addOption("--port", "NUM", "TCP port to listen on.");
        argp.addOption("--bind", "ADDR", "Address to bind to (default: 0.0.0.0).");
        argp.addOption("--staticroot", "PATH",
                "Web root from which to serve static files (/s URLs).");
        argp.addOption("--cachedir", "PATH",
                "Directory under which to cache result of requests.");
        argp.addOption("--worker-threads", "NUM",
                "Number for async io workers (default: cpu * 2).");
        argp.addOption("--async-io", "true|false",
                "Use async NIO (default true) or traditional blocking io");
        argp.addOption("--read-only", "true|false",
                "Set tsd.mode to ro (default false)");
        argp.addOption("--disable-ui", "true|false",
                "Set tsd.core.enable_ui to false (default true)");
        argp.addOption("--disable-api", "true|false",
                "Set tsd.core.enable_api to false (default true)");
        argp.addOption("--backlog", "NUM",
                "Size of connection attempt queue (default: 3072 or kernel"
                        + " somaxconn.");
        argp.addOption("--max-connections", "NUM",
                "Maximum number of connections to accept");
        argp.addOption("--flush-interval", "MSEC",
                "Maximum time for which a new data point can be buffered"
                        + " (default: " + DEFAULT_FLUSH_INTERVAL + ").");
        argp.addOption("--statswport", "Force all stats to include the port");
        CliOptions.addAutoMetricFlag(argp);

        args = CliOptions.parse(argp, args);
        args = null;

        Config config = CliOptions.getConfig(argp);

        if (config.getString("tsd.http.staticroot").isEmpty()) {
            usage(argp, "Missing static root directory", 1);
        }

        if (config.getString("tsd.http.cachedir").isEmpty()) {
            usage(argp, "Missing static root directory", 1);
        }

        if (!config.hasProperty("tsd.network.port")) {
            usage(argp, "Missing network port", 1);
        }

        ServerSocketChannelFactory factory;
        int connections_limit = 0;
        connections_limit = config.getInt("tsd.core.connections.limit");
        if (config.getBoolean("tsd.network.async_io")) {
            int workers = Runtime.getRuntime().availableProcessors() * 2;
            if (config.hasProperty("tsd.network.worker_threads")) {
                workers = config.getInt("tsd.network.worker_threads");
            }

            //创建一个服务端  用一个线程池来处理连接
            Executor executor = Executors.newCachedThreadPool();
            NioServerBossPool bossPool = new NioServerBossPool(executor, 1, new Threads.WorkerThreadNamer());
            NioWorkerPool workerPool = new NioWorkerPool(executor, workers, new Threads.WorkerThreadNamer());

            factory = new NioServerSocketChannelFactory(bossPool, workerPool);
        } else {
            factory = new OioServerSocketChannelFactory(Executors.newCachedThreadPool(), Executors.newCachedThreadPool(), new Threads.PrependThreadNamer());
        }

        StartupPlugin startup = null;
        startup = loadStartupPlugins(config);

        tsdb = new TSDB(config);

        if (startup != null) {
            tsdb.setStartupPlugin(startup);
        }
        tsdb.initializePlugins(true);
        if (config.getBoolean("tsd.storage.hbase.prefetch_meta")) {
            tsdb.preFetchHBaseMeta();
        }

        tsdb.checkNecessaryTablesExist().joinUninterruptibly();

        registerShutdownHook();

        ServerBootstrap server = new ServerBootstrap(factory);

        RpcManager manager = RpcManager.instance(tsdb);

        server.setPipelineFactory(new PipelineFactory(tsdb, manager, connections_limit));
        if (config.hasProperty("tsd.network.backlog")) {
            server.setOption("backlog", config.getInt("tsd.network.backlog"));
        }
        server.setOption("child.tcpNoDelay",
                config.getBoolean("tsd.network.tcp_no_delay"));
        server.setOption("child.keepAlive",
                config.getBoolean("tsd.network.keep_alive"));
        server.setOption("reuseAddress",
                config.getBoolean("tsd.network.reuse_address"));

        InetAddress bindAddress = null;
        if (config.hasProperty("tsd.network.bind")) {
            bindAddress = InetAddress.getByName(config.getString("tsd.network.bind"));
        }
        final InetSocketAddress addr = new InetSocketAddress(bindAddress,
                config.getInt("tsd.network.port"));

        server.bind(addr);

        if (startup != null) {
            startup.setReady(tsdb);
        }
        log.info("Ready to serve on " + addr);
    }

    private static void registerShutdownHook() {
        final class TsdbShutDown extends Thread {
            public TsdbShutDown() {
                super("TSDBShutdown");
            }

            @Override
            public void run() {
                try {
                    if (RpcManager.isInitialized()) {
                        RpcManager.instance(tsdb).shutdown().join();
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * Called to gracefully shutdown the plugin. Implementations should close
     * any IO they have open
     *
     * @return A deferred object that indicates the completion of the request.
     * The {@link Object} has not special meaning and can be {@code null}
     * (think of it as {@code Deferred<Void>}).
     */
    public Deferred<ArrayList<Object>> shutdown() {
        // Clear shared instance.
        INSTANCE.set(null);

        final Collection<Deferred<Object>> deferreds = Lists.newArrayList();


        return Deferred.groupInOrder(deferreds);
    }

    private static StartupPlugin loadStartupPlugins(Config config) {

        StartupPlugin startup = null;

        if (config.getBoolean("tsd.startup.enable")) {
            log.debug("Startup Plugin is Enabled");
            String plugin_path = config.getString("tsd.core.plugin_path");
            String plugin_class = config.getString("tsd.startup.plugin");
            log.debug("Plugin Path: " + plugin_path);
        } else {
            startup = null;
        }
        return startup;
    }
}
